﻿#region 
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data.SqlClient;
using System.Data.Common;
using System.Data;
using EtlComm;
using System.Configuration;
using System.Diagnostics;
using System.Threading;
using System.Collections;

#endregion
namespace ETLHandler
{
    class CSQLLoader : ILoader
    {
        ILog log = LogFactory.CreateLog<ServiceBaseLog>();
        string currenSql = string.Empty;

        #region ILoader 成员
        /// <summary>
        /// 加载数据
        /// </summary>
        /// <param name="result">数据集</param>
        /// <returns>执行成功返回TRUE，否则返回FALSE</returns>
        public bool SetResult(System.Data.DataTable result)
        {
            Debug.Assert(result != null);
            if (result.Rows.Count == 0)
                return true;
            DataFactory factory = new CSQLDataFactory();
            IMap map = factory.CreateMap();

            using (SqlConnection c = new SqlConnection(ip))
            {
                try
                {
                    IDictionary<string, TableKey> mapCol = map.Map(int.Parse(result.TableName));
                    string DestTableName = map.GetTableNameByTaskID(int.Parse(result.TableName));

                    lock (EtlGlobal.LockHt.SyncRoot)
                    {
                        if (!EtlGlobal.LockHt.ContainsKey(DestTableName))
                            EtlGlobal.LockHt.Add(DestTableName, DestTableName);
                    }
                    
                    switch (Action)
                    {
                        case SQLAction.ADD:
                            {
                                int times = 0;
                                lock (EtlGlobal.LockHt[DestTableName])
                                {
                                SQLADD:
                                    try
                                    {
                                        ActionAdd(result, c, mapCol, DestTableName);
                                    }
                                    catch (SqlException ex)
                                    {
                                        if (ex.Number == 2627 || ex.Message.Contains("违反了 PRIMARY KEY 约束"))
                                        {
                                            ActionDelete(result, c, mapCol, DestTableName);
                                            Thread.Sleep(500);
                                            if (times++ < 3)
                                                goto SQLADD;
                                            else throw;
                                            //ActionAdd(result, c, mapCol, DestTableName);
                                        }
                                        else
                                            throw;
                                    }
                                }
                            }
                            break;
                        case SQLAction.DELETE:
                            {
                                ActionDelete(result, c, mapCol, DestTableName);
                            }
                            break;
                        case SQLAction.UPDATE:
                            {
                                lock (EtlGlobal.LockHt[DestTableName])
                                {
                                    ActionUpdate(result, c, mapCol, DestTableName);
                                }
                            }
                            break;
                        case SQLAction.DELADD:
                            {
                                int times = 0;
                                lock (EtlGlobal.LockHt[DestTableName])
                                {
                                SQLREDO:
                                    try
                                    {
                                        ActionDelete(result, c, mapCol, DestTableName);
                                        ActionAdd(result, c, mapCol, DestTableName);
                                    }
                                    catch (SqlException ex)
                                    {
                                        if (ex.Number == 2627 || ex.Message.Contains("违反了 PRIMARY KEY 约束"))
                                        {
                                            Thread.Sleep(500);
                                            if (times++ < 3)
                                                goto SQLREDO;
                                            else throw;
                                        }
                                        else
                                            throw;
                                    }
                                    
                                }
                            }
                            break;
                        case SQLAction.ADDUP:
                            {

                            }
                            break;
                        default:

                            break;
                    }
                    return true;
                }
                catch (System.Exception ex)
                {
                    string errinfo = string.Format("加载操作：\t{0} {1}\n{2}\n", DateTime.Now.ToString(),currenSql,ex.ToString());
                    log.WriteLog(result.Rows[0].Field<string>("StationID"), result.TableName, errinfo, LogType.ERROR);
                    return false;
                }
            }
        }
        /// <summary>
        /// 目标数据库IP
        /// </summary>
        public string ip { set; get; }
        /// <summary>
        /// 执行动作
        /// </summary>
        public SQLAction Action { set; get; }

        #endregion

        #region 私有函数
        /// <summary>
        /// 拼接SQL语句的条件部分
        /// </summary>
        /// <param name="result">源数据</param>
        /// <param name="mapCol">主键列</param>
        /// <returns>拼接好的SQL条件</returns>
        private string PrepareSQLStatement(System.Data.DataTable result, IDictionary<string, TableKey> mapCol)
        {
            var PrimaryKeyColl = (from a in mapCol
                                 where a.Value.IsPrimaryKey == true
                                 select a.Value).ToList();
            StringBuilder sbSql = new StringBuilder();
            //生成只有两个主键的SQL
            if (PrimaryKeyColl.Count <= 2)
            {
                foreach (var pk in PrimaryKeyColl)
                {
                    if (pk.Name.Equals(EtlGlobal.SID, StringComparison.CurrentCultureIgnoreCase))
                    {
                        sbSql.AppendFormat(" and {0} = '{1}' ", pk.Name, result.Rows[0][pk.Name]);
                        continue;
                    }
                    var buf = result.AsEnumerable().Select(f => f.Field<object>(pk.Name).ToString()).ToArray();
                    var bufDist = buf.Distinct();
                    if (bufDist.Count() == 1)
                    {
                        sbSql.AppendFormat(" and {0} = '{1}' ", pk.Name, bufDist.First());
                    }
                    else
                    {
                        StringBuilder sqlparams = new StringBuilder();
                        foreach (var m in bufDist)
                            sqlparams.AppendFormat(",{0}", m);
                        string mysql = sqlparams.ToString().Remove(0, 1);
                        sbSql.AppendFormat(" and {0} in ({1})", pk.Name, mysql);
                    }
                }
                return sbSql.ToString();
            }
            else
            {
                //多主键的SQL
                sbSql.Append(" and ");
                string strformat = string.Empty;
                foreach (var pk in PrimaryKeyColl)
                {
                    if (pk.Name.Equals(EtlGlobal.SID,StringComparison.InvariantCultureIgnoreCase))
                        sbSql.AppendFormat("{0} = '{1}' and ", pk.Name, result.Rows[0][pk.Name]);
                    else
                        strformat += string.Format("+'-'+convert(varchar,{0})", pk.Name);
                }
                sbSql.Append(strformat.Remove(0, 5));
                sbSql.Append(" in (");

                var sqlin = result.AsEnumerable().Select(f =>
                {
                    StringBuilder sb = new StringBuilder();
                    foreach (var pk in PrimaryKeyColl)
                    {
                        if (pk.Name.Equals(EtlGlobal.SID, StringComparison.InvariantCultureIgnoreCase))
                            continue;
                        sb.AppendFormat("-{0}", f.Field<object>(pk.Name).ToString());
                    }
                    return sb.ToString().Remove(0, 1);
                }).ToArray();

                foreach (var x in sqlin)
                {
                    sbSql.AppendFormat("'{0}',", x);
                }
                sbSql.Append("'end')");
                return sbSql.ToString();
            }
        }
        

        /// <summary>
        /// 更新数据
        /// </summary>
        /// <param name="result">待更新的数据表</param>
        /// <param name="c">数据库连接</param>
        /// <param name="mapCol">数据映射</param>
        /// <param name="DestTableName">目标表表名</param>
        private void ActionUpdate(System.Data.DataTable result, SqlConnection c, IDictionary<string, TableKey> mapCol, string DestTableName)
        {
            Debug.Assert(result != null);
            if (result.Rows.Count == 0)
                return;

            //主键列
            var pks = from a in mapCol
                      where a.Value.IsPrimaryKey == true
                      select a.Value;

            IList<DataTable> lstDataTable = Split(result, EtlGlobal.MaxRecord);

            foreach (var t in lstDataTable)
            {
                DataTable destTable = new DataTable();
                SqlDataAdapter ada = null;
                SqlCommandBuilder scbBuilder = null;

                //从目标数据库里选择要更新的记录
                using (SqlCommand sqlcmd = new SqlCommand())
                {
                    sqlcmd.CommandText = currenSql = string.Format("select * from {0} where 1=1 {1}", DestTableName, PrepareSQLStatement(t, mapCol));
                    sqlcmd.Connection = c;
                    sqlcmd.CommandTimeout = 60 * 5;
                    ada = new SqlDataAdapter(sqlcmd);
                    ada.MissingSchemaAction = MissingSchemaAction.AddWithKey;
                    scbBuilder = new SqlCommandBuilder(ada);
                    if (c.State == ConnectionState.Closed)
                        c.Open();

                    ada.Fill(destTable);
                }
                //if (destTable.PrimaryKey.Count() == 0)
                //{
                //    ActionDelete(t, c, mapCol, DestTableName);
                //    ActionAdd(t, c, mapCol, DestTableName);
                //    ada.Dispose();
                //    destTable.Dispose();
                //    Debug.WriteLine("---- delete and add for update success.");
                //    continue;
                //}
                foreach (DataRow dr in t.Rows)
                {
                    IList<object> lstpk = new List<object>();
                    foreach (var pkey in destTable.PrimaryKey)
                        lstpk.Add(dr[pkey.ColumnName]);
                    DataRow kkdr = destTable.Rows.Find(lstpk.ToArray());
                    if (kkdr == null)
                    {
                        //把StationID转换到最前面来
                        int drcount = dr.ItemArray.Count();
                        object[] temp = new object[drcount];
                        temp[0] = dr.ItemArray[drcount - 1];
                        Array.Copy(dr.ItemArray, 0, temp, 1, drcount - 1);

                        destTable.Rows.Add(temp);
                        continue;
                    }
                    foreach (DataColumn x in destTable.Columns)
                        if (!destTable.PrimaryKey.Contains(x))
                            kkdr[x.ColumnName] = dr[x.ColumnName];
                }
                ada.Update(destTable);
                ada.Dispose();
                destTable.Dispose();
            }
            
            c.Close();
            log.WriteLog(string.Format("---{0} {1} UPDATE 操作成功.",DateTime.Now.ToString(),currenSql));
        }
        /// <summary>
        /// 添加数据
        /// </summary>
        /// <param name="result">等插入的数据表</param>
        /// <param name="c">数据库连接</param>
        /// <param name="mapCol">数据映射</param>
        /// <param name="DestTableName">目标表表名</param>
        private void ActionAdd(System.Data.DataTable result, SqlConnection c, IDictionary<string, TableKey> mapCol, string DestTableName)
        {
            Debug.Assert(result != null);
            if (result.Rows.Count == 0)
                return;
            if (c.State == ConnectionState.Closed)
                c.Open();
            using (SqlBulkCopy DesBulkOp = new SqlBulkCopy(c, SqlBulkCopyOptions.UseInternalTransaction, null))
            {
                DesBulkOp.DestinationTableName = DestTableName;
                DesBulkOp.BulkCopyTimeout = 60 * 50;
                foreach (var objmap in mapCol)
                {
                    DesBulkOp.ColumnMappings.Add(objmap.Key, objmap.Value.Name);
                }
                DesBulkOp.WriteToServer(result);
            }
            c.Close();
        }
        /// <summary>
        /// 删除数据
        /// </summary>
        /// <param name="result">待删除的数据</param>
        /// <param name="c">数据库连接</param>
        /// <param name="mapCol">数据映射</param>
        /// <param name="DestTableName">目标表表名</param>
        private void ActionDelete(DataTable result, SqlConnection c, IDictionary<string, TableKey> mapCol, string DestTableName)
        {
            Debug.Assert(result != null);
            if (result.Rows.Count == 0)
                return;
            if (c.State == ConnectionState.Closed)
                c.Open();
            
            using (SqlCommand sqlcmd = new SqlCommand())
            {
                sqlcmd.CommandTimeout = 60 * 5;
                sqlcmd.Connection = c;
                sqlcmd.CommandText = string.Format("select count(*) from {0} where {1}= '{2}'", DestTableName, EtlGlobal.SID, result.Rows[0][EtlGlobal.SID]);
                int rnum = (int)sqlcmd.ExecuteScalar();
                if (rnum == result.Rows.Count)
                {
                    sqlcmd.CommandText = string.Format("delete from {0} where {1}= '{2}'", DestTableName, EtlGlobal.SID, result.Rows[0][EtlGlobal.SID]);
                    sqlcmd.ExecuteNonQuery();
                    Debug.WriteLine("---- " + sqlcmd.CommandText);
                    return;
                }
            }
            IList<DataTable> lstDataTable = Split(result,EtlGlobal.MaxRecord);
            foreach (var t in lstDataTable)
            {
                using (SqlCommand sqlcmd = new SqlCommand())
                {
                    sqlcmd.CommandTimeout = 60 * 5;
                    sqlcmd.Connection = c;
                    sqlcmd.CommandText = string.Format("delete from {0} where 1=1 {1}", DestTableName, PrepareSQLStatement(t, mapCol));
                    int delCount = sqlcmd.ExecuteNonQuery();
                    Debug.WriteLine(string.Format("---- recoder:{0} delete:{1} {2}",t.Rows.Count,delCount,sqlcmd.CommandText));
                }
            }
            c.Close();
        }

        /// <summary>
        /// 把result拆分为几个小的dataTable
        /// </summary>
        /// <param name="result"></param>
        /// <returns></returns>
        private IList<DataTable> Split(DataTable result,int maxRecord)
        {
            IList<DataTable> lstDataTable = new List<DataTable>();
            for (int i = 0; i <= result.Rows.Count / maxRecord; i++)
            {
                DataSet ds = new DataSet();
                ds.Merge(result.AsEnumerable().Skip(i * maxRecord).Take(maxRecord).ToArray());
                if (ds.Tables.Count > 0)
                    lstDataTable.Add(ds.Tables[0]);
            }
            return lstDataTable;
        }

        

        #endregion
    }
}
